package com.shujia.streaming

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.ReceiverInputDStream

import java.sql.{Connection, DriverManager, PreparedStatement}


object Demo8DS2Mysql {
  def main(args: Array[String]): Unit = {
    //使用DataFrame的语法
    val sparkSession: SparkSession = SparkSession.builder()
      .master("local[2]")
      .appName("rdd与DStream的关系")
      .config("spark.sql.shuffle.partitions", "1")
      .getOrCreate()
    import org.apache.spark.sql.functions._
    import sparkSession.implicits._

    //使用RDD的语法
    val sparkContext: SparkContext = sparkSession.sparkContext

    //使用DStream的语法
    val streamingContext = new StreamingContext(sparkContext, Durations.seconds(5))
    val infoDS: ReceiverInputDStream[String] = streamingContext.socketTextStream("master", 10086)


    //    infoDS.foreachRDD((rdd:RDD[String])=>{
    //      println("======================= 正在处理一批数据 ==========================")
    //      //处理rdd中每一条数据
    //      rdd.foreach((line:String)=>{
    //        //如果将创建连接的代码写在这里，这样的话，每条数据都会创建一次连接
    //        /**
    //         * 创建与数据库连接对象
    //         */
    //        //注册驱动
    //        Class.forName("com.mysql.jdbc.Driver")
    //        //创建数据库连接对象
    //        val conn: Connection = DriverManager.getConnection(
    //          "jdbc:mysql://master:3306/bigdata30?useUnicode=true&characterEncoding=UTF-8&useSSL=false",
    //          "root",
    //          "123456"
    //        )
    //        //创建预编译对象
    //        val statement: PreparedStatement = conn.prepareStatement("insert into students values(?,?,?,?,?)")
    //
    //        val info: Array[String] = line.split(",")
    //        statement.setInt(1,info(0).toInt)
    //        statement.setString(2,info(1))
    //        statement.setInt(3,info(2).toInt)
    //        statement.setString(4,info(3))
    //        statement.setString(5,info(4))
    //
    //        //执行sql语句
    //        statement.executeUpdate()
    //
    //        //释放资源
    //        statement.close()
    //        conn.close()
    //      })
    //    })


    /**
     * 设想中的改造，不一定能运行
     * 我们将原本在rdd中创建连接的代码放到了ds中，发现PreparedStatement不能与task任务一起序列化到executor中的
     * 这样的写法是不可以的！！！！！！！
     */
    //    infoDS.foreachRDD((rdd: RDD[String]) => {
    //      println("======================= 正在处理一批数据 ==========================")
    //      //如果将创建连接的代码写在这里，这样的话，每条数据都会创建一次连接
    //      /**
    //       * 创建与数据库连接对象
    //       */
    //      //注册驱动
    //      Class.forName("com.mysql.jdbc.Driver")
    //      //创建数据库连接对象
    //      val conn: Connection = DriverManager.getConnection(
    //        "jdbc:mysql://master:3306/bigdata30?useUnicode=true&characterEncoding=UTF-8&useSSL=false",
    //        "root",
    //        "123456"
    //      )
    //      //创建预编译对象
    //      val statement: PreparedStatement = conn.prepareStatement("insert into students values(?,?,?,?,?)")
    //      //处理rdd中每一条数据
    //      rdd.foreach((line: String) => {
    //        val info: Array[String] = line.split(",")
    //        statement.setInt(1, info(0).toInt)
    //        statement.setString(2, info(1))
    //        statement.setInt(3, info(2).toInt)
    //        statement.setString(4, info(3))
    //        statement.setString(5, info(4))
    //        //执行sql语句
    //        statement.executeUpdate()
    //      })
    //
    //      //释放资源
    //      statement.close()
    //      conn.close()
    //
    //    })

    /**
     * rdd中有一个算子foreachPartition
     * rdd本质是由一系列分区构成的，如果我们可以将分区数设置为1，且每个分区创建一个连接不就好了么
     */
    infoDS.foreachRDD((rdd: RDD[String]) => {
      println("======================= 接收到 5s 一批次数据 ==========================")
      rdd.repartition(2)
      println(s" DS封装的RDD中的分区数为:${rdd.getNumPartitions} ")

      /**
       * foreachPartition，处理一个分区的数据
       * 将一个分区的数据，封装成了一个迭代器
       */
      rdd.foreachPartition((itr: Iterator[String]) => {
        println("======================= 正在处理一个分区的数据 ==========================")
        //如果将创建连接的代码写在这里，这样的话，每条数据都会创建一次连接
        /**
         * 创建与数据库连接对象
         */
        //注册驱动
        Class.forName("com.mysql.jdbc.Driver")
        //创建数据库连接对象
        val conn: Connection = DriverManager.getConnection(
          "jdbc:mysql://master:3306/bigdata30?useUnicode=true&characterEncoding=UTF-8&useSSL=false",
          "root",
          "123456"
        )
        //创建预编译对象
        val statement: PreparedStatement = conn.prepareStatement("insert into students values(?,?,?,?,?)")
        println("========================= 创建了一次连接 =========================")
        itr.foreach((line: String) => {
          val info: Array[String] = line.split(",")
          statement.setInt(1, info(0).toInt)
          statement.setString(2, info(1))
          statement.setInt(3, info(2).toInt)
          statement.setString(4, info(3))
          statement.setString(5, info(4))
          //执行sql语句
          statement.executeUpdate()
        })
        statement.close()
        conn.close()
      })
    })


    streamingContext.start()
    streamingContext.awaitTermination()
    streamingContext.stop()
  }
}
